[REVIEW NOTIFICATION]

This pull request has been approved by:

• asddongmen
• nongfushanquan

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

In addition to these tests, I also tested the multi-topic feature.

1. Start Kop with docker-compose
2. Start TiDB cluster tiup playground v6.5.0 --tiflash 0
3. Build and start TiCDC server
4. Create a changefeed with multi-topic config: ./cdc cli changefeed create --sink-uri="kafka://10.2.xxx:19092/testdsada?max-message-bytes=1048576&protocol=canal-json&kafka-version=2.9.0&replication-factor=1" --config=./changefeed.toml

[sink]
enable-kafka-sink-v2 = true
dispatchers = [{matcher = ['*.*'], topic = "tidb_{schema}_{table}"}]

5. Write some data in upstream

mysql> use test;
Database changed
mysql> CREATE TABLE example1 (   id INT PRIMARY KEY,   b INT );
Query OK, 0 rows affected (0.15 sec)

mysql> INSERT INTO example1 (id,b) VALUES (2,2);
Query OK, 1 row affected (0.01 sec)

mysql> CREATE TABLE example2 (   id INT PRIMARY KEY,   b INT );
Query OK, 0 rows affected (0.09 sec)

mysql> INSERT INTO example2 (id,b) VALUES (2,2);
Query OK, 1 row affected (0.00 sec)

7. check the topic with kcat:

Metadata for all topics (from broker 1960160947: 10.2.xxxx:19092/1960160947):
 1 brokers:
  broker 1960160947 at 10.2.xxxx7:19092 (controller)
 4 topics:
  topic "testdsada" with 3 partitions:
    partition 0, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 1, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 2, leader 1960160947, replicas: 1960160947, isrs: 1960160947
  topic "tidb_test_example2" with 3 partitions:
    partition 0, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 1, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 2, leader 1960160947, replicas: 1960160947, isrs: 1960160947
  topic "not_exist" with 3 partitions:
    partition 0, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 1, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 2, leader 1960160947, replicas: 1960160947, isrs: 1960160947
  topic "tidb_test_example1" with 3 partitions:
    partition 0, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 1, leader 1960160947, replicas: 1960160947, isrs: 1960160947
    partition 2, leader 1960160947, replicas: 1960160947, isrs: 1960160947

8. Check the data with kcat:

kcat  -b 10.2.xxxx:19092 -t tidb_test_example2 -o -10000
% Auto-selecting Consumer mode (use -P or -C to override)
{"id":0,"database":"test","table":"example2","pkNames":null,"isDdl":true,"type":"CREATE","es":1683358998100,"ts":1683358998598,"sql":"CREATE TABLE `example2` (`id` INT PRIMARY KEY,`b` INT)","sqlType":null,"mysqlType":null,"data":null,"old":null}
{"id":0,"database":"test","table":"example2","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1683359005400,"ts":1683359005922,"sql":"","sqlType":{"id":4,"b":4},"mysqlType":{"id":"int","b":"int"},"old":null,"data":[{"id":"2","b":"2"}]}


kcat  -b 10.2.xxxx:19092 -t tidb_test_example1 -o -10000
% Auto-selecting Consumer mode (use -P or -C to override)
{"id":0,"database":"test","table":"example1","pkNames":null,"isDdl":true,"type":"CREATE","es":1683358986400,"ts":1683358986592,"sql":"CREATE TABLE `example1` (`id` INT PRIMARY KEY,`b` INT)","sqlType":null,"mysqlType":null,"data":null,"old":null}
{"id":0,"database":"test","table":"example1","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1683358992000,"ts":1683358992389,"sql":"","sqlType":{"id":4,"b":4},"mysqlType":{"id":"int","b":"int"},"old":null,"data":[{"id":"2","b":"2"}]}

/test all

I tested that both "Standalone" and "Cluster" work.

1. Standalone: https://github.com/streamnative/kop/blob/master/docs/kop.md#run-kop-on-standalone-pulsar-in-docker-compose
2. Cluster: https://github.com/streamnative/kop/blob/master/docs/kop.md#run-kop-in-pulsar-with-component-for-each-system-using-docker-compose

/retest

/test all

/retest

/merge

/test all

This pull request has been accepted and is ready to merge.

/test all

/test pull-dm-integration-test

@hi-rustin: The specified target(s) for /test were not found.
The following commands are available to trigger required jobs:

• /test cdc-integration-kafka-test
• /test cdc-integration-mysql-test
• /test cdc-integration-storage-test
• /test dm-compatibility-test
• /test dm-integration-test
• /test engine-integration-test
• /test verify

Use /test all to run the following jobs that were automatically triggered:

• pingcap/tiflow/ghpr_verify

/test dm-integration-test

/test all

/retest

In response to a cherrypick label: new pull request created to branch release-6.1: #8904.

In response to a cherrypick label: new pull request created to branch release-6.5: #8905.

In response to a cherrypick label: new pull request created to branch release-7.1: #8906.

/cherry-pick release-6.1-20230516-v6.1.3

@hi-rustin: new pull request created to branch release-6.1-20230516-v6.1.3: #8966.